package process;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.List;

/**
 * 定时器使用
 */
public class ProcessingTimeTimerDemo {

    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        DataStreamSource<String> lines = env.socketTextStream("localhost", 8888);

        SingleOutputStreamOperator<Tuple2<String, Integer>> mapped = lines.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String s) throws Exception {
                String[] fields = s.split(",");
                return Tuple2.of(fields[0], Integer.parseInt(fields[1]));
            }
        });

        KeyedStream<Tuple2<String, Integer>, String> keyed = mapped.keyBy(f -> f.f0);

        keyed.process(new KeyProcessTimer()).print();


        env.execute("");
    }

    private static class KeyProcessTimer extends KeyedProcessFunction<String, Tuple2<String, Integer>, Tuple2<String, Integer>> {

        // 定义状态
        private transient ValueState<List<Tuple2<String,Integer>>> state;


        @Override
        public void open(Configuration parameters) throws Exception {
            // 创建状态描述器
            ValueStateDescriptor<List<Tuple2<String, Integer>>> stateDescriptor = new ValueStateDescriptor<>("lst-state", TypeInformation.of(new TypeHint<List<Tuple2<String, Integer>>>() {
            }));
            // 获取状态
            state = getRuntimeContext().getState(stateDescriptor);
        }

        @Override
        public void processElement(Tuple2<String, Integer> value, Context context, Collector<Tuple2<String, Integer>> out) throws Exception {
            // 获取状态
            List<Tuple2<String, Integer>> lst = state.value();
            if (lst == null){
                lst = new ArrayList<Tuple2<String,Integer>>();
            }
            // 添加数据
            lst.add(value);
            // 更新状态
            state.update(lst);
            // 注册定时器
            context.timerService().registerProcessingTimeTimer(1646104800000L);
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<Tuple2<String, Integer>> out) throws Exception {
            for (Tuple2<String, Integer> e : state.value()) {
                out.collect(e);
            }
        }
    }

}
